/*
 * Copyright (C) 2014 Qloudgen Technology
*/



import org.apache.log4j.PropertyConfigurator;

import com.qloudgen.csb.client.Connection;
import com.qloudgen.csb.client.Consumer;
import com.qloudgen.csb.client.KfkConsumer;
import com.qloudgen.csb.client.KfkMessageListener;
import com.qloudgen.csb.client.KfkProducer;
import com.qloudgen.csb.client.Message;
import com.qloudgen.csb.client.MessageListener;
import com.qloudgen.csb.client.Queue;
import com.qloudgen.csb.client.Session;

public class KfkEndpoint implements KfkMessageListener {  
	
	long count = 0;
	KfkConsumer kc;
	KfkProducer kp;
	
	private KfkEndpoint(String topic, String group, String zkConn,
			int bufferSize) {
		kc = new KfkConsumer(topic, group, zkConn, bufferSize, null);
		kc.setMessageListener(this);
		
		kp = new KfkProducer(zkConn, -1, null);
	}
	
	@Override
	public void onMessage(String message, long offset) {
		//System.out.println(message);
		//try {
		//	Thread.sleep(2);
		//} catch (InterruptedException e) {
		//	e.printStackTrace();
		//}
		count++;
		if (count >= 1000) {
			System.out.println(count + " messages processed...");
			count = 0;
		}
	}

    public static void main(String[] args) throws Exception {
    	PropertyConfigurator.configure("conf/log4j.properties");
    	System.out.println("providing...");
    	
    	try {
    		//kc.run();
    		//new Thread(new Provider("tcp://" + args[0], null)).start();
    	} catch (Exception e) {
    		e.printStackTrace();
    	}
    	
    	System.out.println("running...");
    	//System.exit(0);
    }
}

